package com.rocketmq.controller;

import com.alibaba.fastjson.JSON;
import com.rocketmq.entity.OrderEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

@RestController
@Slf4j
public class ProducerController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @RequestMapping("/sendMsg")
    public String sendMsg() {
        OrderEntity orderEntity = new OrderEntity(123456,"发送消息");
        // 第一种方式：普通发送
        rocketMQTemplate.convertAndSend("myTest", JSON.toJSONString(orderEntity));
        // 第二种方式：封装消息体发送
        /*Message<String> message = new GenericMessage<>("hi zhengzhou");
        rocketMQTemplate.convertAndSend("myTest",message);*/
        
        return "success";
    }
    @RequestMapping("/sendSyncMsg")
    public String sendSyncMsg() {
        OrderEntity orderEntity = new OrderEntity(123456, "同步发送消息");
        for (int i = 0; i < 10; i++) {
            // 发送同步消息
            SendResult myTest = rocketMQTemplate.syncSend("myTest", JSON.toJSONString(orderEntity));
            System.out.println(i + "-" + myTest.getSendStatus());
        }
        return "success";
    }

    @RequestMapping("/sendASyncMsg")
    public String sendASyncMsg() {
        OrderEntity orderEntity = new OrderEntity(123456, "异步发送消息");
        for (int i = 0; i < 10; i++) {
            int index = i;
            rocketMQTemplate.asyncSend("myTest", JSON.toJSONString(orderEntity), new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    //成功回调
                    log.info(index+"-"+sendResult.getSendStatus());
                }
                @Override
                public void onException(Throwable e) {
                    //异常回调
                    log.info(e.getMessage());
                }
            });
        }
        return "success";
    }
    /**
     * 发送单向消息
     * @param topic
     * @param msg
     */
    public void sendOneWayMsg(String topic, String msg){
        this.rocketMQTemplate.sendOneWay(topic,msg);
    }
    /**************验证rocketmq顺序消费***************/
    @RequestMapping("/sendOrderedMsg")
    public String sendOrderedMsg(){
        /**
         * hashKey: 为了保证报到同一个队列中，将消息发送到orderTopic主题上
         */
        rocketMQTemplate.syncSendOrderly("orderTopic","no1","order");
        rocketMQTemplate.syncSendOrderly("orderTopic","no2","order");
        rocketMQTemplate.syncSendOrderly("orderTopic","no3","order");
        rocketMQTemplate.syncSendOrderly("orderTopic","no4","order");
        return "success";
    }

    /**
     * 发送延迟消息
     * 消息内容为json格式
     */
    @RequestMapping("/sendDelayMsg")
    public String sendAsyncMsgByJsonDelay() throws RemotingException, InterruptedException, MQClientException {
        OrderEntity orderEntity = new OrderEntity(123456, "异步延迟消息");

        //消息内容将orderExt转为json
        String json = JSON.toJSONString(orderEntity);
        Message message =
                new Message("delayTopic",json.getBytes(StandardCharsets.UTF_8));
        //设置延迟等级
        message.setDelayTimeLevel(4);
        //发送异步消息
        this.rocketMQTemplate.getProducer().send(message,new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("sendResult"+sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                log.error(throwable.getMessage());
            }
        });

        System.out.printf("send msg : %s",orderEntity);
        return "success";
    }
    @RequestMapping("/sendBatch")
    public String sendBatch() {
        List<Integer> ids = List.of(1001, 1002, 1003, 1004, 1005);
        // <1>  创建多条 Demo02Message 消息
        List<org.springframework.messaging.Message> messages = new ArrayList<>(ids.size());
        for (long id : ids) {
            // 创建 Demo02Message 消息
            OrderEntity message = new OrderEntity();
            message.setId(id);
            // 构建 Spring Messaging 定义的 Message 消息
            messages.add(MessageBuilder.withPayload(message).build());
        }
        rocketMQTemplate.syncSend("batchTopic", messages, 30 * 1000L);
        return "success";
    }
}